Changes in this file look like a refactoring rather than part of the feature, can this be extracted in a separate PR?

same here, an unrelated (though maybe tempting) refactoring…

This code is needed by the new code, but isn't available when written as a protected method. The protected method looks like it can be removed because it is only used in one place, but I left it since removing it could break any subclasses that override it. I could mark it as @deprecated instead.

I think it is unnecessary to separate them with semicolon and prefixing should be enough. My problem with semicolon separation is that with many enough configs it makes the list a little bit hard to interpret.

Ok. I can switch the semicolons to commas and provide an example that demonstrates tidy formatting. In case I've misunderstood your request, the reason for the separator character is that I want to provide a mechanism for users to add any number of key=value pairs to the list rather than mandating which configs are permitted.

Sorry I can't switch the semicolons to commas because individual values can have commas. I can make sure the docs make it clear that readable formatting is easy and desirable.

All default configs should be part of cruisecontrol.properties and we should avoid hard-coding (similarly to goals and default.goals). This would simplify your code as well as you wouldn't need to merge configs.

If you only prefixed the configs, then you could get them the following way in an array and avoid duplicating functionality with getMap:

config.originalsWithPrefix("persisted.data.kafka.topic."),
config.originalsWithPrefix("persisted.data.kafka.producer."),
config.originalsWithPrefix("persisted.data.kafka.consumer."),

Ok, makes sense. Thanks for the examples.

If I understand correctly, you do this because this way we can serialize demoted or removed brokers with their timestamp the same way in both cases (in-memory vs Kafka).
To be completely honest, I find this a bit less intuitive to understand at first. For this I'd like to propose that what if we move the type of serialization to the respective storage method? For instance we could have a class like this:

BrokerEvent implements Serializable {
  private String brokerId;
  private long timestamp;
  private EventType eventType; // demote or remove
  private Namespace namespace; // currently "executor" only
}

For memory serialization we could use the standard ByteArrayInput/OutputStream serialization while for Kafka topics we could use json that is serialized into Strings. That would make the Kafka topics easily debuggable too.
When deserialized, we can represent them in a Map<String, BrokerEvent> structure to map broker ID to the event to make it easier to query.
Let me know your thoughts.